package org.jetlinks.pro.timeseries.query;

import lombok.AllArgsConstructor;
import org.hswebframework.web.dict.EnumDict;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.math.MathFlux;

import java.util.Comparator;
import java.util.function.Function;

@AllArgsConstructor
public enum Aggregation implements EnumDict<String> {

    MIN(numberFlux -> MathFlux.min(numberFlux, Comparator.comparing(Number::doubleValue))),
    MAX(numberFlux -> MathFlux.max(numberFlux, Comparator.comparing(Number::doubleValue))),
    AVG(numberFlux -> MathFlux.averageDouble(numberFlux, Number::doubleValue)),
    SUM(numberFlux -> MathFlux.sumDouble(numberFlux, Number::doubleValue)),
    COUNT(Flux::count),
    FIRST(numberFlux -> numberFlux.take(1).single()),
    TOP(numberFlux -> numberFlux.take(1).single()),

    MEDIAN(numberFlux -> org.jetlinks.pro.utils.math.MathFlux.median(numberFlux, Number::doubleValue)),//中位数
    SPREAD(numberFlux -> Mono.error(new UnsupportedOperationException("SPREAD unsupported"))),//差值
    STDDEV(numberFlux -> org.jetlinks.pro.utils.math.MathFlux.standardDeviation(numberFlux, Number::doubleValue)),//标准差

    NONE(numberFlux -> Mono.just(0));

    private final Function<Flux<Number>, Mono<? extends Number>> computer;


    public <S> Mono<? extends Number> compute(Flux<S> source, Function<S, Number> mapper) {

        return computer.apply(source.map(mapper));
    }

    @Override
    public String getValue() {
        return name();
    }

    @Override
    public String getText() {
        return name();
    }

    @Override
    public boolean isWriteJSONObjectEnabled() {
        return false;
    }
}